package com.ce.socket;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;

public class ASocketPool extends BsSocketPool implements ISocketPool {
	private static Logger log = Logger.getLogger(ASocketPool.class);
	private DaemonThread daemonThread;
	private Map<ISocket,Long> availPool;	//可用的连接
	private Map<ISocket,Long> busyPool;	//正在使用的连接
	private Map<ISocket,Integer> deadPool;	//需要回收的连接
	//仅用于测试
	public Map<ISocket,Long> totalPool = new HashMap<ISocket, Long>();;
	public ASocketPool(String server) {
		super(server);
	}

	public ISocket getSocket() {
		// TODO Auto-generated method stub
		return null;
	}

	public void initialize() {
		synchronized( this ) {

			// check to see if already initialized
			if ( initialized
					&& ( availPool != null )
					&& ( busyPool != null ) ) {
				log.error( "-> trying to initialize an already initialized pool" );
				return;
			}

			// pools
			availPool   = new HashMap<ISocket,Long>( initConn );
			busyPool    = new HashMap<ISocket,Long>( initConn );
			deadPool    = new IdentityHashMap<ISocket,Integer>();

			log.debug( "-> initializing pool with following settings:" );
			log.debug( "-> initial size: " + initConn );
			log.debug( "-> min spare   : " + minConn );
			log.debug( "-> max spare   : " + maxConn );

			// if servers is not set, or it empty, then
			// throw a runtime exception
			if ( server == null || server.length() <= 0 ) {
				log.error( "-> server is illegal" );
				throw new IllegalStateException( "++++ server is illegal" );
			}

			// initalize our internal hashing structures
			populateBuckets();
			
			// mark pool as initialized
			this.initialized = true;

			// start maint thread
			if ( this.daemonSleep > 0 )
				this.startDaemonThread();
		}

	}
	private void populateBuckets() {
		log.debug( "-> initializing internal hashing structure for consistent hashing" );

		// create initial connections
		log.debug( "-> creating initial connections (" + initConn + ") for host: " + server );

		for ( int j = 0; j < initConn; j++ ) {
			ISocket socket = createSocket();
			if ( socket == null ) {
				log.error( "-> failed to create connection to: " + server + " -- only " + j + " created." );
				break;
			}

			addSocketToPool( availPool, socket );
		}
	}
	private ISocket createSocket() {

		ISocket socket = null;

		try {
			socket = new ASocket( server, this.socketTO, this.socketConnectTO, this.nagle );
			totalPool.put(socket, new Long( System.currentTimeMillis() ));
			if ( !socket.isConnected() ) {
				log.error( "-> failed to get DalSocket obj for: " + server + " -- new socket is not connected" );
				deadPool.put( socket, ZERO );
				socket = null;
			}
			log.info( "-> created and added socket: " + socket.toString() + " for host " + server );
		}
		catch ( Exception ex ) {
			log.error( "-> failed to create DalSocket obj for: " + server +" "+ ex);
			socket = null;
		}
		
		return socket;
	}
	private synchronized void checkAllSocket() {
		if(aliveCheck) {
			boolean flag = false;
			for ( Iterator<ISocket> j = availPool.keySet().iterator(); j.hasNext(); ) {
				ISocket socket = j.next();
				boolean isAlive = socket.isAlive();
				flag = flag || isAlive;
				if(!isAlive) {
					deadPool.put( socket, ZERO );
					j.remove();
				}
				log.debug( "-> "+socket+" is status:"+ isAlive);
			}
			Set<ISocket> aliveSocket = availPool.keySet();
			for (ISocket socket : aliveSocket) {
				//更新有效连接的活跃时间
				addSocketToPool(availPool, socket);
			}
			poolIsAvail = flag;
		}
	}
	protected void addSocketToPool( Map<ISocket,Long> pool, ISocket socket ) {
		if ( pool != null )
			pool.put( socket, new Long( System.currentTimeMillis() ) );
	}
	
	public byte[] sendData(byte[] data, int retimes) throws Exception {
		ISocket socket = null;
		try {
			if(retimes == 0) {
				log.info("-> retry send data end, failed! retimes:"+retimes);
				throw new DALException("send data failed! retimes:"+retimes);
			}
			socket = getConnection();
			socket.write(data);
			socket.flush();
			if(retimes < 3)
				log.info("-> retry send data end, succeed! retimes:"+retimes);
			byte[] brep = socket.read();
			recoverySocket(socket, true);
			return brep;
		} catch (DALException e) {
			log.error("-> send time:"+retimes+" error:" , e);
			throw e;
		} catch (Exception e) {
			log.error("-> send time:"+retimes+" error:" , e);
			if(socket != null) {
				recoverySocket(socket, false);
				checkAllSocket();
			}
			return sendData(data, retimes-1);
		}
	}
	protected void recoverySocket( ISocket socket, boolean addToAvail) {
		if(socket == null)
			return;
		log.debug( "-> calling check-in on socket: " + socket.toString() );

		synchronized( this ) {
			// remove from the busy pool
			log.debug( "-> removing socket (" + socket.toString() + ") from busy pool");
			removeSocketFromPool( busyPool, socket );
			
			if ( socket.isConnected() && addToAvail ) {
				// add to avail pool
				log.debug( "-> returning socket (" + socket.toString() + " to avail pool");
				addSocketToPool( availPool, socket );
			}
			else {
				deadPool.put( socket, ZERO );
				socket = null;
			}
			log.info( "->recoverySocket avail/busy/dead/poolIsAvail:("+availPool.size()+"/"+busyPool.size()+"/"+deadPool.size()+"/"+poolIsAvail+")");
		}
	}
	private void removeSocketFromPool( Map<ISocket,Long> pool, ISocket socket ) {
		if ( pool != null )
			pool.remove( socket );
	}
	public ISocket getConnection() {

		if ( !this.initialized ) {
			log.error( "-> attempting to get SockIO from uninitialized pool!" );
			return null;
		}
		log.info( "->getConnection-1 avail/busy/dead/poolIsAvail:("+availPool.size()+"/"+busyPool.size()+"/"+deadPool.size()+"/"+poolIsAvail+")");		
		synchronized( this ) {
			log.info( "->getConnection-2 avail/busy/dead/poolIsAvail:("+availPool.size()+"/"+busyPool.size()+"/"+deadPool.size()+"/"+poolIsAvail+")");
			// if we have items in the pool
			// then we can return it
			if ( availPool != null && !availPool.isEmpty() ) {
				for ( Iterator<ISocket> i = availPool.keySet().iterator(); i.hasNext(); ) {
					ISocket socket = i.next();
					if ( socket.isConnected() ) {
						log.debug( "-> moving socket for host (" + server + ") to busy pool ... socket: " + socket );
						// remove from avail pool
						i.remove();

						// add to busy pool
						addSocketToPool( busyPool, socket );

						// return socket
						return socket;
					}
					else {
						// add to deadpool for later reaping
						deadPool.put( socket, ZERO );

						// remove from avail pool
						i.remove();
					}
				}
			} else {
				if(busyPool != null && busyPool.size() >= maxConn) {
					log.info("-> availPool("+availPool.size()+") is empty, and busyPool("+busyPool.size()+") more than "+ maxConn);
					return null;
				}
			}
			// create one socket -- let the maint thread take care of creating more
			ISocket socket = createSocket();
			if ( socket != null ) {
				addSocketToPool( busyPool, socket );
				return socket;
			}
		}
		return null;
	}
	
	private void startDaemonThread() {

		if ( daemonThread != null ) {

			if ( daemonThread.isRunning() ) {
				log.error( "=== daemon thread already running ===" );
			}
			else {
				daemonThread.start();
			}
		}
		else {
			daemonThread = new DaemonThread( this );
			daemonThread.setInterval( this.daemonSleep );
			daemonThread.start();
		}
	}
	private void stopDaemonThread() {
		if ( daemonThread != null && daemonThread.isRunning() )
			daemonThread.stopThread();
	}
	public void shutDown() {
		synchronized( this ) {
			log.debug( "=== SockIOPool shutting down... ===" );

			if ( daemonThread != null && daemonThread.isRunning() ) {
				// stop the main thread
				stopDaemonThread();

				// wait for the thread to finish
				while ( daemonThread.isRunning() ) {
					log.debug( "-> waiting for daemon thread to finish run +++" );
					try { Thread.sleep( 500 ); } catch ( Exception ex ) { }
				}
			}

			log.debug( "-> closing all internal pools." );
			closePool( availPool );
			closePool( busyPool );
			availPool         = null;
			busyPool          = null;
			daemonThread       = null;
			initialized       = false;
			log.debug( "=== SockIOPool finished shutting down. ===" );
		}
	}

	private void closePool( Map<ISocket,Long> pool ) {
		 for ( Iterator<ISocket> j = pool.keySet().iterator(); j.hasNext(); ) {
			 ISocket socket = j.next();

			 try {
				 socket.close();
			 }
			 catch ( IOException ioe ) {
				 log.error( "-> failed to trueClose socket: " + socket.toString() + " for host: " + server );
			 }

			 j.remove();
			 socket = null;
		 }
	}

	private void selfDaemon() {
		log.debug( "=== Starting self maintenance.... ===" );

		Integer needSockets = 0;

		synchronized( this ) {

			log.info( "->selfDaemon-1 avail/busy/dead/poolIsAvail:("+availPool.size()+"/"+busyPool.size()+"/"+deadPool.size()+"/"+poolIsAvail+")");

			// if pool is too small (n < minSpare)
			if ( availPool.size() + busyPool.size() < minConn ) {
				// need to create new sockets
				needSockets = minConn - availPool.size() - busyPool.size();
			}
		}

		log.debug( "-> Need to create " + needSockets + " new sockets for pool for host: " + server );

		Set<ISocket> newSock = new HashSet<ISocket>( needSockets );
		for ( int j = 0; j < needSockets; j++ ) {
			ISocket socket = createSocket();

			if ( socket == null )
				break;

			newSock.add( socket );
		}


		// synchronize to add and remove to/from avail pool
		// as well as clean up the busy pool (no point in releasing
		// lock here as should be quick to pool adjust and no
		// blocking ops here)
		int needToClose = 0;
		synchronized( this ) {
			for ( ISocket socket : newSock )
				addSocketToPool( availPool, socket );

			log.info( "->selfDaemon-1 avail/busy/dead/poolIsAvail:("+availPool.size()+"/"+busyPool.size()+"/"+deadPool.size()+"/"+poolIsAvail+")");

			if ( availPool.size() + busyPool.size() > maxConn ) {
				// need to close down some sockets
				int diff        = availPool.size() + busyPool.size() - maxConn;
				needToClose = (diff <= poolMultiplier) ? diff : (diff) / poolMultiplier;

				log.debug( "-> need to remove " + needToClose + " spare sockets for pool for host: " + server );

				for ( Iterator<ISocket> j = availPool.keySet().iterator(); j.hasNext(); ) {
					if ( needToClose <= 0 )
						break;

					// remove stale entries
					ISocket socket = j.next();
					deadPool.put( socket, ZERO );
					j.remove();
					needToClose--;
				}
			} else {
				for ( Iterator<ISocket> j = availPool.keySet().iterator(); j.hasNext(); ) {
					// remove stale entries
					ISocket socket = j.next();
					long expire   = availPool.get( socket ).longValue();

					// if past idle time
					// then close socket
					// and remove from pool
					if ( (expire + maxIdle) < System.currentTimeMillis() ) {
						log.debug( "-> removing stale entry from pool as it is past its idle timeout and pool is over max spare" );

						// remove from the availPool
						deadPool.put( socket, ZERO );
						j.remove();
					}
				}
			}
		}

		// go through busy sockets and destroy sockets
		// as needed to maintain pool settings

		log.debug( "-> Size of busy pool for host (" + server + ")  = " + busyPool.size() );

		// loop through all connections and check to see if we have any hung connections
		for ( Iterator<ISocket> j = busyPool.keySet().iterator(); j.hasNext(); ) {
			// remove stale entries
			ISocket socket = j.next();
			long hungTime = busyPool.get( socket ).longValue();

			// if past max busy time
			// then close socket
			// and remove from pool
			if ( (hungTime + maxBusyTime) < System.currentTimeMillis() ) {
				log.error( "-> removing potentially hung connection from busy pool ... socket in pool for " + (System.currentTimeMillis() - hungTime) + "ms" );

				// remove from the busy pool
				deadPool.put( socket, ZERO );
				j.remove();
			}
		}
		//check all socket in availPool whether avail
		checkAllSocket();
		
		log.info( "->selfDaemon-2 avail/busy/dead/poolIsAvail:("+availPool.size()+"/"+busyPool.size()+"/"+deadPool.size()+"/"+poolIsAvail+")");
		// finally clean out the deadPool
		Set<ISocket> toClose;
		synchronized( deadPool ) {
			toClose  = deadPool.keySet();
			deadPool = new IdentityHashMap<ISocket,Integer>();
		}

		for ( ISocket socket : toClose ) {
			try {
				socket.close();
			}
			catch ( Exception ex ) {
				log.error( "-> failed to close DalSocket obj from deadPool " , ex);
			}

			socket = null;
		}

		log.debug( "=== ending self maintenance. ===" );
	}
	private static class DaemonThread extends Thread {

		// logger
		private static Logger log = Logger.getLogger( DaemonThread.class );

		private ASocketPool pool;
		private long interval      = 1000 * 3; // every 3 seconds
		private boolean stopThread = false;
		private boolean running;

		protected DaemonThread( ASocketPool pool ) {
			this.pool = pool;
			this.setDaemon( true );
			this.setName( "MaintThread" );
		}

		public void setInterval( long interval ) { this.interval = interval; }
		
		public boolean isRunning() {
			return this.running;
		}

		/** 
		 * sets stop variable
		 * and interupts any wait 
		 */
		public void stopThread() {
			this.stopThread = true;
			this.interrupt();
		}

		/** 
		 * Start the thread.
		 */
		public void run() {
			this.running = true;

			while ( !this.stopThread ) {
				try {
					Thread.sleep( interval );

					// if pool is initialized, then
					// run the maintenance method on itself
					if ( pool.isInitialized() )
						pool.selfDaemon();

				}
				catch ( Exception e ) {
					log.error("-> selfMaint error " + e);
				}
			}

			this.running = false;
		}
	}
	public void check() {
		// TODO Auto-generated method stub
		
	}

	public void setParam(Object... args) {
		// TODO Auto-generated method stub
	}
}
